package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/7/19 16:06
 */
public class Dwd_03_DwdTradeCartAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_03_DwdTradeCartAdd().init(
            2003,
            2,
            "Dwd_03_DwdTradeCartAdd"
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 1. 建立动态表读取ods_db数据
        readOdsDb(tEnv, "Dwd_03_DwdTradeCartAdd");
        
        // 2. 读取 base_dic的数据
        readBaseDic(tEnv);
        
        // 3. 从ods_db中过滤出加购数据
        Table cartInfo = tEnv.sqlQuery("select " +
                                           " `data`['id'] id, " +
                                           " `data`['user_id'] user_id, " +
                                           " `data`['sku_id'] sku_id, " +
                                           " `data`['cart_price'] cart_price, " +
                                           " if(`type`='insert', " +
                                           "     `data`['sku_num'], " +
                                           "   cast(cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int) as string)) sku_num, " +
                                           " `data`['source_type'] source_type_code, " +
                                           " `data`['source_id'] source_id, " +
                                           " ts, " +
                                           " pt " +
                                           "from ods_db " +
                                           "where `database`='gmall2022' " +
                                           "and `table`='cart_info' " +
                                           "and (`type`='insert' " +
                                           "    or(`type`='update' " +
                                           "        and `old`['sku_num'] is not null " +  // 这次的更新是由于sku_num变化引起
                                           "        and cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int)" +
                                           "      )" +
                                           ")");
        tEnv.createTemporaryView("cart_info", cartInfo);
        
        // 4. 加购和base_dic进 lookup join , 内连接
        Table result = tEnv.sqlQuery("select " +
                                        " ci.id, " +
                                        " ci.user_id, " +
                                        " ci.sku_id, " +
                                        " ci.source_id, " +
                                        " ci.source_type_code, " +
                                        " bd.dic_name source_type_name, " +
                                        " ci.sku_num, " +
                                        " ci.ts " +
                                        "from cart_info ci " +
                                        "join base_dic for system_time as of ci.pt as bd " +
                                        "on ci.source_type_code=bd.dic_code");
        
        // 5. join后的结果写入到Kafka中
        tEnv.executeSql("create table dwd_trade_cart_add(" +
                            "id string," +
                            "user_id string," +
                            "sku_id string," +
                            "source_id string," +
                            "source_type_code string," +
                            "source_type_name string," +
                            "sku_num string," +
                            "ts string " +
                            ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CART_ADD));
        
        result.executeInsert("dwd_trade_cart_add");
                        
        
    }
}
